package com.rrd.dw.mr.gzip;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.HadoopUtils;
import com.rrd.dw.utils.MD5Util;

public class CreditCardBillMR extends Configured implements Tool {
    public static final String FILED_SPLIT_FLAG = "\001";
    public static final int SIZEOF_MD5 = 16;
    public static final byte[] INF_FAMILY = Bytes.toBytes("inf");
    public static final byte[] JOB_FAMILY = Bytes.toBytes("job");

    public static class ExportCreditCardMapper extends TableMapper<Text, Text> {
        private String[] arr = null;
        private Map<String, String> map = new HashMap<String, String>();
        private JSONObject carBillObj = null;
        private Text valText = new Text();
        private Text keyText = new Text();
        private Object[] idnumberObj = null;

        @Override
        protected void map(ImmutableBytesWritable key, Result value,
                Context context) throws IOException, InterruptedException {
            map.clear();
            // 身份证号
            idnumberObj = decodeRowkey(key.get());
            if (idnumberObj != null && idnumberObj[1] != null) {
                map.put("id_number", idnumberObj[1].toString());
            }

            // 判断账单字段是否为null
            if (value.getValue(INF_FAMILY,
                Bytes.toBytes("card_bill_report")) == null) {
                return;
            }
            carBillObj = JSON.parseObject(Bytes.toString(
                value.getValue(INF_FAMILY, Bytes.toBytes("card_bill_report"))));
            if (carBillObj == null) {
                return;
            }

            String start_time = value.getValue(JOB_FAMILY,
                Bytes.toBytes("start_time")) == null ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(JOB_FAMILY,
                        Bytes.toBytes("start_time"))));

            map.put("product",
                value.getValue(JOB_FAMILY, Bytes.toBytes("system_id")) == null
                    ? ""
                    : Bytes.toString(value.getValue(JOB_FAMILY,
                        Bytes.toBytes("system_id"))));
            map.put("start_time", start_time.replace(FILED_SPLIT_FLAG, ""));
            map.put("job_id",
                value.getValue(JOB_FAMILY, Bytes.toBytes("job_id")) == null ? ""
                    : Bytes.toString(
                        value.getValue(JOB_FAMILY, Bytes.toBytes("job_id"))));

            parseCardBillReport(context);

        }

        public void parseCardBillReport(Context context)
                throws IOException, InterruptedException {
            JSONObject tmpObj = null;
            JSONArray jsonData = null;
            this.map.put("email", this.carBillObj.getString("email"));
            StringBuilder sb = new StringBuilder();
            String id = "";
            if (carBillObj.getJSONArray("data") != null) {
                jsonData = carBillObj.getJSONArray("data");
                for (int i = 0; i < jsonData.size(); i++) {
                    tmpObj = jsonData.getJSONObject(i).getJSONObject("body");
                    if (tmpObj == null) {
                        continue;
                    }
                    sb.delete(0, sb.length());
                    sb.append(map.get("id_number") == null ? ""
                        : map.get("id_number"));
                    sb.append(
                        map.get("product") == null ? "" : map.get("product"));
                    sb.append(map.get("email") == null ? "" : map.get("email"));
                    sb.append(
                        map.get("job_id") == null ? "" : map.get("job_id"));
                    sb.append(tmpObj.getString("bank_code") == null ? ""
                        : tmpObj.getString("bank_code"));
                    sb.append(tmpObj.getString("card_no") == null ? ""
                        : tmpObj.getString("card_no"));
                    sb.append(tmpObj.getString("bill_date") == null ? ""
                        : tmpObj.getString("bill_date"));
                    sb.append(tmpObj.getString("start_time") == null ? ""
                        : tmpObj.getString("start_time"));
                    sb.append(i);
                    id = MD5Util.getMd5(sb.toString());
                    sb.delete(0, sb.length());
                    sb.append(id).append(FILED_SPLIT_FLAG);
                    sb.append(
                        map.get("product") == null ? "" : map.get("product"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(map.get("email") == null ? "" : map.get("email"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(map.get("id_number") == null ? ""
                        : map.get("id_number")).append(FILED_SPLIT_FLAG);
                    sb.append(
                        map.get("job_id") == null ? "" : map.get("job_id"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(map.get("start_time") == null ? ""
                        : map.get("start_time")).append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "bank_code"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "bank_name"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "card_no"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "credit_limit"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "last_bill_balance"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "bill_balance"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "bill_amount_due"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "bill_date"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "min_payment"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "adjustment"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "payment_due_date"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "integral"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "interest"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "last_amount_due"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "cash_limit"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "is_original"))
                        .append(FILED_SPLIT_FLAG);
                    sb.append(getJsonVal(tmpObj, "email_sender"));
                    // sb.append(FILED_SPLIT_FLAG).append(tmpObj.toJSONString());
                    this.valText.set(sb.toString());
                    this.keyText.set(id);
                    context.write(this.keyText, this.valText);
                    parseDetails(context, id, tmpObj.getJSONArray("details"));
                }
            }

        }

        private void parseDetails(Context context, String parent_id,
                JSONArray arr) throws IOException, InterruptedException {

            JSONObject tmpObj = null;
            StringBuilder sb = new StringBuilder();
            if (arr == null || arr.size() <= 0) {
                return;
            }
            for (int i = 0; i < arr.size(); i++) {
                tmpObj = arr.getJSONObject(i);
                sb.delete(0, sb.length());
                sb.append(parent_id).append(i).append(FILED_SPLIT_FLAG);
                sb.append(parent_id).append(FILED_SPLIT_FLAG);
                sb.append(getJsonVal(tmpObj, "transaction_date  "))
                    .append(FILED_SPLIT_FLAG);
                sb.append(getJsonVal(tmpObj, "amount", "0"))
                    .append(FILED_SPLIT_FLAG);
                sb.append(getJsonVal(tmpObj, "currency"))
                    .append(FILED_SPLIT_FLAG);
                sb.append(getJsonVal(tmpObj, "description"))
                    .append(FILED_SPLIT_FLAG);
                sb.append(getJsonVal(tmpObj, "transaction_area"));
                this.valText.set(sb.toString());
                this.keyText.set(parent_id);
                context.write(this.keyText, this.valText);
            }

        }

        private String getJsonVal(JSONObject obj, String key) {
            return obj.getString(key) == null ? "" : obj.getString(key);
        }

        private String getJsonVal(JSONObject obj, String key, String val) {
            return obj.getString(key) == null ? val : obj.getString(key);
        }

    }

    public static class ExportCreditCardReducer
            extends Reducer<Text, Text, Text, Text> {
        private MultipleOutputs<Text, Text> mos;
        private NullWritable key2 = NullWritable.get();

        @Override
        protected void setup(Context context)
                throws IOException, InterruptedException {
            this.mos = new MultipleOutputs<Text, Text>(context);
        }

        @Override
        protected void cleanup(Context context)
                throws IOException, InterruptedException {
            this.mos.close();
        }

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            for (Text val : values) {
                if (val.toString().split(FILED_SPLIT_FLAG).length > 7) {
                    this.mos.write("bills", key2, val);
                } else {
                    this.mos.write("details", key2, val);
                }
            }

        }
    }

    public static Object[] decodeRowkey(byte[] raw) {
        if (raw.length < SIZEOF_MD5 + Bytes.SIZEOF_LONG) {
            return null;
        }
        Object[] ret = new Object[2];
        ret[0] = Long.MAX_VALUE
            - Bytes.toLong(raw, SIZEOF_MD5, Bytes.SIZEOF_LONG);

        if (raw.length > SIZEOF_MD5 + Bytes.SIZEOF_LONG) {
            ret[1] = Bytes.toString(raw, SIZEOF_MD5 + Bytes.SIZEOF_LONG);
        }

        return ret;
    }

    @Override
    public int run(String[] args) throws Exception {
        Properties prop = new Properties();
        try (InputStream is = this.getClass().getClassLoader()
            .getResourceAsStream("hbase.properties");) {
            prop.load(is);
        }
        Job job = Job.getInstance(HBaseConfiguration.create(this.getConf()));
        Configuration conf = job.getConfiguration();
        conf.set("hbase.zookeeper.quorum",
            prop.get("hbase.zookeeper.quorum").toString());
        conf.set("hbase.rootdir", prop.get("hbase.rootdir").toString());

        conf.set("mapreduce.map.memory.mb", "3072");
        conf.set("mapreduce.reduce.memory.mb", "5120");
        conf.set("hbase.client.scanner.timeout.period", "120");
        conf.set("mapreduce.output.fileoutputformat.compress", "true");
        conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
        conf.set("mapreduce.output.fileoutputformat.compress.codec",
            "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("io.compression.codecs",
            "org.apache.hadoop.io.compress.GzipCodec");
        String table = "credit_card_bills";
        String outDir = conf.get("param.out.path");
        System.out.println("开始日期：" + conf.get("param.start_date"));
        System.out.println("结束日期：" + conf.get("param.stop_date"));
        System.out.println("输入hbase表：" + table);
        System.out.println("输出路径：" + outDir);
        Long startTime = DateUtil
            .formateToTimestamp(conf.get("param.start_date"));
        Long endTime = DateUtil.formateToTimestamp(conf.get("param.stop_date"));

        // String isTest=conf.get("param.isTest","0");
        Path outpath = new Path(outDir);

        HadoopUtils.delete(conf, outpath);
        job.setJarByClass(CreditCardBillMR.class);
        List<Scan> scans = new ArrayList<Scan>();
        Scan scan = new Scan();
        scan.setCacheBlocks(false);
        scan.addColumn(INF_FAMILY, Bytes.toBytes("card_bill_report"));
        scan.addColumn(JOB_FAMILY, Bytes.toBytes("system_id"));
        scan.addColumn(JOB_FAMILY, Bytes.toBytes("start_time"));
        scan.addColumn(JOB_FAMILY, Bytes.toBytes("job_id"));
        scan.setCaching(200);
        scan.setMaxVersions(1);
        scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME,
            Bytes.toBytes(table));
        scan.setTimeRange(startTime, endTime + 14400000);
        // scan.setStartRow(Bytes.toBytes(""));
        // scan.getStopRow(Bytes.toBytes(""));
        scans.add(scan);

        TableMapReduceUtil.initTableMapperJob(scans,
            ExportCreditCardMapper.class, Text.class, Text.class, job);
        FileOutputFormat.setOutputPath(job, outpath);
        MultipleOutputs.addNamedOutput(job, "bills", TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, "details", TextOutputFormat.class,
            Text.class, Text.class);

        job.setReducerClass(ExportCreditCardReducer.class);
        // job.setNumReduceTasks(1);
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        int res = ToolRunner.run(conf, new CreditCardBillMR(), args);
        System.exit(res);
    }

}
